use core::hash;
use std::{collections::HashMap, fs, path::Path, str::FromStr, sync::Arc};

use crate::ftp::FtpDownload;

use super::{
    errors::{Error, Result},
    get_ch_pool, get_columns, get_dict_tables, get_work_path, is_exist_sql, search_calc_ref_join,
};
use arc_swap::ArcSwapOption;
use chrono::NaiveDate;
use clickhouse_rs::types::SqlType;
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use strum::IntoEnumIterator;
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct Config {
    #[serde(default)]
    pub frequency: Frequency,
    pub file_name_prefix: String,
    pub file_ext_name: String,
    pub confirm_file_ext_name: String,
    pub column_separator: u8,
    pub tables: Vec<Table>,
    #[serde(default)]
    pub coded_format: CodedFormat,
    pub ftp: Ftp,
}
impl Config {
    pub async fn parse(
        &mut self,
        source_type: &SourceType,
        dicts: &Vec<(String, String)>,
    ) -> Result<()> {
        //验证tables和ftp
        //所有的来自clickhouse数据源的DICT表，必须在reload_dict_tables中定义，根据DICT表的定义生成reload_dict_tables
        for table in &mut self.tables {
            table.parse().await.or_else(|err| {
                Err(Error::from_imp_conf(
                    source_type.to_owned(),
                    &table.table_name,
                    err.to_string(),
                ))
            })?;
            for (dn, tn) in dicts.iter() {
                if table.reload_dict_tables.contains(dn) {
                    continue;
                }
                let is_match = if tn == &table.table_name {
                    true
                } else {
                    if tn.starts_with("V_") {
                        let tn1 = &tn[2..];
                        let tn2 = &table.table_name[2..];
                        tn1.contains(tn2) || tn2.contains(tn1)
                    } else {
                        false
                    }
                };
                if is_match {
                    table.reload_dict_tables.push(dn.to_string());
                }
            }
        }
        self.ftp.parse().or_else(|err| {
            Err(Error::from_imp_conf(
                source_type.to_owned(),
                "ftp",
                err.to_string(),
            ))
        })?;
        Ok(())
    }
}

#[derive(
    Debug,
    Clone,
    Copy,
    hash::Hash,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Deserialize,
    Serialize,
    strum::EnumString,
    strum::Display,
    strum::EnumIter,
)]
pub enum SourceType {
    ODS,
    EDW,
    FTP,
    CAP,
    DLA,
    EDA,
}
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize, strum::EnumString, strum::Display,
)]
pub enum CodedFormat {
    UTF8,
    GBK,
}
impl Default for CodedFormat {
    fn default() -> Self {
        CodedFormat::UTF8
    }
}
#[derive(
    Debug, Clone, PartialEq, Eq, Deserialize, Serialize, strum::EnumString, strum::Display,
)]
pub enum EtlStage {
    TxtParse,
    Import,
    DiviProc,
    AcblProc,
    BigTable,
}
#[derive(
    Debug, Clone, PartialEq, Eq, Deserialize, Serialize, strum::EnumString, strum::Display,
)]
pub enum Frequency {
    #[strum(serialize = "day")]
    Daily,
    #[strum(serialize = "montj")]
    Monthly,
    #[strum(serialize = "quarter")]
    Quarterly,
}
impl Default for Frequency {
    fn default() -> Self {
        Frequency::Daily
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct Ftp {
    pub ip: String,
    pub port: u16,
    pub user_name: String,
    pub password: String,
    pub path: String,
}
impl Ftp {
    pub fn parse(&self) -> anyhow::Result<()> {
        let mut ftp_conn = FtpDownload::from_conf(&self)?;
        ftp_conn.close()?;
        Ok(())
    }
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct Table {
    pub table_name: String,
    pub file_name: String,
    pub import_type: ImportType,
    pub is_must: bool,
    #[serde(default)]
    pub can_blank: bool,
    pub columns: HashMap<String, u32>,
    #[serde(default)]
    pub unique_keys: Vec<String>,
    #[serde(default)]
    pub rs_calc_cols: HashMap<String, String>,
    #[serde(default)]
    pub with_cause: Option<String>,
    #[serde(default)]
    pub db_calc_cols: HashMap<String, String>,
    #[serde(default)]
    pub file_name_suffix: String,
    #[serde(default)]
    pub file_sn: String,
    #[serde(default)]
    pub acbl_col_names: Vec<String>,
    #[serde(default)]
    pub divi_fact_convs: Vec<DiviFactConv>,
    #[serde(default)]
    pub ignore_error_cols: Vec<String>,
    #[serde(default)]
    pub acbl_procs: Option<AcblProc>,
    #[serde(default)]
    pub before_sqls: Option<SqlList>,
    #[serde(default)]
    pub after_sqls: Option<SqlList>,
    #[serde(default)]
    pub start_row_no: u8,
    #[serde(default)]
    pub reload_dict_tables: Vec<String>,
    #[serde(default)]
    pub dependents: Vec<Dependent>,
}
impl Table {
    pub async fn parse(&mut self) -> anyhow::Result<()> {
        //验证table_name和columns是否存在，
        let cols = get_columns(&self.table_name)
            .await
            .or_else(|err| Err(anyhow::anyhow!("表{}不存在，{}", &self.table_name, err)))?;
        for col in self.columns.keys() {
            if !cols.iter().any(|(c, _)| c == col) {
                return Err(anyhow::anyhow!(
                    "columns配置，表{}中没有列{}",
                    self.table_name,
                    col
                ));
            }
        }
        //calc_cols配置的字段名是否存在
        for col in self.rs_calc_cols.keys() {
            if !cols.iter().any(|(c, _)| c == col) {
                return Err(anyhow::anyhow!(
                    "columns配置，表{}中没有列{}",
                    self.table_name,
                    col
                ));
            }
        }
        let has_incr = self.db_calc_cols.values().any(|v| v == "#INCR");
        if self.acbl_procs.is_some() {
            if self.import_type != ImportType::InsertAndUpdate {
                return Err(anyhow::anyhow!(
                    "表{}中定义了acbl_procs，但import_type不是InsertAndUpdate",
                    self.table_name
                ));
            }
            if !has_incr {
                return Err(anyhow::anyhow!(
                    "表{}中定义了acbl_procs，但没有发现#INCR列",
                    self.table_name
                ));
            }
        }
        if has_incr && self.unique_keys.len() == 0 {
            return Err(anyhow::anyhow!(
                "表{}rs_calc_cols中定义了#INCR列，但unique_keys为空",
                self.table_name
            ));
        }
        for col in self.db_calc_cols.keys() {
            if !cols.iter().any(|(c, _)| c == col) {
                return Err(anyhow::anyhow!(
                    "columns配置，表{}中没有列{}",
                    self.table_name,
                    col
                ));
            }
        }
        //unique_keys,acbl_col_names配置的字段名是否存在
        for col in self.unique_keys.iter() {
            if !cols.iter().any(|(c, _)| {
                if col.starts_with("toString(") {
                    let col = &col[9..col.len() - 1];
                    return c == col;
                }
                c == col
            }) {
                return Err(anyhow::anyhow!(
                    "columns配置，表{}中没有列{}",
                    self.table_name,
                    col
                ));
            }
        }
        for col in self.acbl_col_names.iter() {
            if !cols.iter().any(|(c, _)| c == col) {
                return Err(anyhow::anyhow!(
                    "acbl_col_names配置，表{}中没有列{}",
                    self.table_name,
                    col
                ));
            }
        }
        if self.import_type == ImportType::MonthTrls {
            if self.acbl_col_names.len() == 0 {
                return Err(anyhow::anyhow!(
                    "表{}中定义了import_type为MonthTrls，acbl_col_names不能为空",
                    self.table_name
                ));
            }
        }
        //验证acbl_procs,字段是否存在
        if let Some(ap) = &self.acbl_procs {
            if self.acbl_col_names.len() == 0 {
                return Err(anyhow::anyhow!(
                    "表{}中定义了acbl_procs，但acbl_col_names为空",
                    self.table_name
                ));
            }
            for col in ap.col_map.values() {
                if !cols.iter().any(|(c, _)| {
                    if col.starts_with("#") {
                        return true;
                    }
                    if col.starts_with("'") && col.ends_with("'") {
                        return true;
                    }
                    if col.contains("(") && col.ends_with(")") {
                        return true;
                    }
                    c == col
                }) {
                    return Err(anyhow::anyhow!(
                        "acbl_procs-col_map配置，表{}中没有列{}",
                        self.table_name,
                        col
                    ));
                }
            }
            ap.parse().await?; //解析acbl_procs
        }
        //生成dependents
        if let Some(wc) = &self.with_cause {
            let deps = search_calc_ref_join(wc);
            for (st, fname) in deps.iter() {
                let is_ex = self
                    .dependents
                    .iter()
                    .any(|d| d.source == **st && d.file_name == *fname);
                if is_ex {
                    continue;
                }
                if *fname==self.file_name {
                    continue;
                }
                self.dependents.push(Dependent {
                    source: st.to_owned().to_owned(),
                    file_name: fname.to_string(),
                    is_along: false,
                });
            }
        }
        for (k, v) in self.rs_calc_cols.iter() {
            if v.trim().is_empty() {
                return Err(anyhow::anyhow!("rs_calc_cols配置，列{k}值不能为空"));
            }
        }
        let trls_divi_key_col = if (self.import_type == ImportType::DayTrls
            || self.import_type == ImportType::MonthTrls)
            && self.divi_fact_convs.len() > 0
        {
            Some(self.divi_fact_convs[0].key_col_name.as_str())
        } else {
            None
        };
        let mut trls_divi_ok=false;
        for (k, v) in self.db_calc_cols.iter() {
            if v.trim().is_empty() {
                return Err(anyhow::anyhow!("db_calc_cols配置，列{k}值不能为空"));
            }
            if let Some(kc)=trls_divi_key_col {
                if kc==k && v.ends_with(kc){
                    trls_divi_ok=true
                }
            }
        }
        if trls_divi_key_col.is_some() && !trls_divi_ok {
            let kc=trls_divi_key_col.unwrap();
            return Err(anyhow::anyhow!("db_calc_cols的配置中必须有列{kc}，并且别名设置为{kc}"));
        }
        //生成dependents
        self.db_calc_cols.values().for_each(|v| {
            let deps = search_calc_ref_join(v);
            for (st, fname) in deps.iter() {
                let is_ex = self
                    .dependents
                    .iter()
                    .any(|d| d.source == **st && d.file_name == *fname);
                if is_ex {
                    continue;
                }
                if *fname==self.file_name {
                    continue;
                }
                self.dependents.push(Dependent {
                    source: st.to_owned().to_owned(),
                    file_name: fname.to_string(),
                    is_along: false,
                });
            }
        });
        //验证divi_fact_convs
        let dfc_len = self.divi_fact_convs.len();
        for dfc in self.divi_fact_convs.iter() {
            if dfc.to_dept_public && !cols.iter().any(|(c, _)| c == &dfc.dept_col) {
                return Err(anyhow::anyhow!(
                    "divi_fact_convs-dept_col配置，表{}中没有列{}",
                    self.table_name,
                    &dfc.dept_col
                ));
            }
            if dfc_len > 1 {
                if dfc.where_col_name.is_none() || dfc.where_col_val.is_none() {
                    return Err(anyhow::anyhow!("divi_fact_convs配置{dfc_len}个，两个及以上必须定义where_col_name和where_col_val"));
                }
            }
        }
        if let Some(sl) = &self.before_sqls {
            let res = match sl {
                SqlList::Vec(vs) => {
                    if vs.iter().any(|s| s.contains(";")) {
                        Err(anyhow::anyhow!("before_sqls配置，不能包含分号;"))
                    } else {
                        Ok(())
                    }
                }
                SqlList::Path(n) => {
                    if !is_exist_sql(sl) {
                        Err(anyhow::anyhow!("before_sqls配置，文件{n}不存在"))
                    } else {
                        Ok(())
                    }
                }
            };
            res?;
        }
        Ok(())
    }
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct Dependent {
    pub source: SourceType,
    pub file_name: String,
    #[serde(default)]
    pub is_along: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
#[serde(untagged)]
pub enum SqlList {
    Vec(Vec<String>),
    Path(String),
}
impl SqlList {
    pub fn from_file_name(s: &str) -> Self {
        if s.ends_with(".sql") {
            SqlList::Path(s.to_string())
        } else {
            SqlList::Path(format!("{}.sql", s))
        }
    }
}
#[derive(
    Debug, Clone, PartialEq, Eq, Deserialize, Serialize, strum::EnumString, strum::Display,
)]
pub enum ImportType {
    InsertAndUpdate,
    TruncateAndInsert,
    InsertOnly,
    DayTrls,
    MonthTrls,
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
pub struct AcblProc {
    pub acbl_table_name: String,
    pub col_map: HashMap<String, String>,
    pub having_cause: Option<String>,
}
impl AcblProc {
    pub async fn parse(&self) -> anyhow::Result<()> {
        //验证acbl_table_name是否存在，
        let cols = get_columns(&self.acbl_table_name).await?;
        //col_map配置的字段名是否存在
        for col in self.col_map.keys() {
            if !cols.iter().any(|(c, _)| c == col) {
                return Err(anyhow::anyhow!("表{}中没有列{}", self.acbl_table_name, col));
            }
        }
        //abl表中必须有DTID日期列，SIGN列，DIVI列
        if !cols.iter().any(|(c, t)| c == "DTID" && *t == SqlType::Date) {
            return Err(anyhow::anyhow!(
                "表{}中必须有列DTID，并且类型必须为Date",
                self.acbl_table_name
            ));
        }
        if !cols.iter().any(|(c, t)| c == "SIGN" && *t == SqlType::Int8) {
            return Err(anyhow::anyhow!(
                "表{}中必须有列SIGN，并且类型必须为UInt8",
                self.acbl_table_name
            ));
        }
        let has_divi_dpid = cols.iter().any(|(c, t)| {
            if let SqlType::Array(t) = t {
                c == "DIVI.DPID" && *t == &SqlType::UInt32
            } else {
                false
            }
        });
        if !has_divi_dpid {
            return Err(anyhow::anyhow!(
                "表{}中必须有列DIVI.DPID，并且类型必须为SqlType::UInt32",
                self.acbl_table_name
            ));
        }
        let has_divi_emid = cols.iter().any(|(c, t)| {
            if let SqlType::Array(t) = t {
                c == "DIVI.EMID" && *t == &SqlType::UInt32
            } else {
                false
            }
        });
        if !has_divi_emid {
            return Err(anyhow::anyhow!(
                "表{}中必须有列DIVI.EMID，并且类型必须为SqlType::UInt32",
                self.acbl_table_name
            ));
        }
        let has_divi_dsvl = cols.iter().any(|(c, t)| {
            if let SqlType::Array(t) = t {
                c == "DIVI.DSVL" && *t == &SqlType::Float64
            } else {
                false
            }
        });
        if !has_divi_dsvl {
            return Err(anyhow::anyhow!(
                "表{}中必须有列DIVI.DSVL，并且类型必须为SqlType::Float64",
                self.acbl_table_name
            ));
        }
        Ok(())
    }
}
#[derive(
    Debug,
    Clone,
    Copy,
    PartialEq,
    Eq,
    PartialOrd,
    Deserialize,
    Serialize,
    strum::EnumIter,
    strum::EnumString,
    strum::Display,
)]
pub enum DiviType {
    #[strum(serialize = "comp_acco")]
    CompAcco,
    #[strum(serialize = "indi_save")]
    IndiSave,
    #[strum(serialize = "indi_loan")]
    IndiLoan,
    #[strum(serialize = "comp_biz_fee")]
    CompBizFee,
    #[strum(serialize = "indi_biz_fee")]
    IndiBizFee,
    #[strum(serialize = "indi_clie")]
    IndiClie,
}
impl DiviType {
    pub async fn drop_table(&self, date: &NaiveDate) -> anyhow::Result<()> {
        let tname = self.join_table_name();
        if tname.is_empty() {
            return Ok(());
        }
        let sql_str = format!(
            "drop table if exists {tname}_{}",
            date.format("%Y%m%d").to_string()
        );
        let mut client = get_ch_pool().get_handle().await?;
        client.execute(sql_str).await?;
        Ok(())
    }
    pub fn acco_table_name(&self) -> &'static str {
        match self {
            DiviType::CompAcco => "T_COMP_ACCO",
            DiviType::IndiSave => "T_INDI_ACCO",
            DiviType::IndiLoan => "T_INDI_LOAN",
            DiviType::CompBizFee => "T_CM_BIZ_FEE_SUM_M",
            DiviType::IndiBizFee => "T_CM_BIZ_FEE_SUM_M",
            DiviType::IndiClie => "",
        }
    }
    pub fn divi_table_name(&self) -> &'static str {
        match self {
            DiviType::CompAcco => "T_COMP_ACCO_DIVI",
            DiviType::IndiSave => "T_INDI_ACCO_DIVI",
            DiviType::IndiLoan => "T_INDI_LOAN_DIVI",
            DiviType::CompBizFee => "T_COMP_ACCO_DIVI",
            DiviType::IndiBizFee => "T_INDI_ACCO_DIVI",
            DiviType::IndiClie => "T_INDI_CLIE_YGFC",
        }
    }
    pub fn join_table_name(&self) -> &'static str {
        match self {
            DiviType::CompAcco => "tmp_JOIN_COMP_ACCO_DIVI",
            DiviType::IndiSave => "tmp_JOIN_INDI_ACCO_DIVI",
            DiviType::IndiLoan => "tmp_JOIN_INDI_LOAN_DIVI",
            DiviType::CompBizFee | DiviType::IndiBizFee => "",
            DiviType::IndiClie => "tmp_JOIN_INDI_CLIE_DIVI",
        }
    }
    pub fn back_table_name(&self) -> &'static str {
        match self {
            DiviType::CompAcco => "JOIN_COMP_ACCO_DIVI_BACK",
            DiviType::IndiSave => "JOIN_INDI_ACCO_DIVI_BACK",
            DiviType::IndiLoan => "JOIN_INDI_LOAN_DIVI_BACK",
            DiviType::CompBizFee | DiviType::IndiBizFee => "JOIN_CM_BIZ_FEE_DIVI_BACK",
            DiviType::IndiClie => "JOIN_INDI_ACCO_DIVI_BACK",
        }
    }
    pub fn back_dict_name(&self) -> &'static str {
        match self {
            DiviType::CompAcco => "DICT_COMP_ACCO_DIVI_BACK",
            DiviType::IndiSave => "DICT_INDI_ACCO_DIVI_BACK",
            DiviType::IndiLoan => "DICT_INDI_LOAN_DIVI_BACK",
            DiviType::CompBizFee | DiviType::IndiBizFee => "",
            DiviType::IndiClie => "",
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Deserialize, Serialize)]
pub struct DiviFactConv {
    pub divi_type: DiviType,
    pub key_col_name: String,
    #[serde(default)]
    pub to_dept_public: bool,
    #[serde(default = "default_dept_col")]
    pub dept_col: String,
    #[serde(default)]
    pub where_col_name: Option<String>,
    #[serde(default)]
    pub where_col_val: Option<String>,
}
fn default_dept_col() -> String {
    "UTID".to_string()
}
static CONF: Lazy<ArcSwapOption<HashMap<SourceType, &'static Config>>> =
    Lazy::new(|| ArcSwapOption::const_empty());

pub async fn init_imp_config() -> Result<()> {
    let conf_path = format!("{}/config", get_work_path());
    let mut conf_map: HashMap<SourceType, &'static Config> = HashMap::new();
    let dicts = get_dict_tables().await?;
    for st in SourceType::iter() {
        let file_path = format!(
            "{}/{}_import.yaml",
            conf_path,
            st.to_string().to_lowercase()
        );
        let conf_str = fs::read_to_string(&file_path)?;
        let mut conf: Config = serde_yaml::from_str(&conf_str)
            .or_else(|err| Err(anyhow::anyhow!("配置文件{}解析失败:{}", &file_path, err)))?;
        conf.parse(&st, &dicts).await?;
        let conf_ref = Box::leak(Box::new(conf));
        conf_map.insert(st, conf_ref);
    }
    CONF.store(Some(Arc::new(conf_map)));
    let no_dicts = get_no_reload_dicts(&dicts);
    if no_dicts.len() > 0 {
        return Err(anyhow::anyhow!(
            "存在没有配置到table-reload_dict_tables的字典表: {:?}",
            no_dicts
        )
        .into());
    }
    Ok(())
}
pub async fn reload_imp_config(path: &Path) -> Result<()> {
    let file_name = path.file_name().unwrap().to_str().unwrap();
    let st_name = file_name[0..file_name.len() - 11].to_uppercase();
    let st = SourceType::from_str(&st_name)
        .or_else(|_| Err(anyhow::anyhow!("无效的SourceType:{st_name}")))?;
    let conf_str = fs::read_to_string(path)?;
    let dicts = get_dict_tables().await?;
    let mut conf: Config = serde_yaml::from_str(&conf_str).or_else(|err| {
        Err(anyhow::anyhow!(
            "配置文件{}解析失败:{}",
            path.to_str().unwrap(),
            err
        ))
    })?;
    conf.parse(&st, &dicts).await?;
    let conf_ref = Box::leak(Box::new(conf));
    let conf_map = CONF.load().clone().unwrap();
    let mut conf_map2: HashMap<SourceType, &'static Config> = HashMap::new();
    for st2 in SourceType::iter() {
        if st2 == st {
            conf_map2.insert(st2, conf_ref);
        } else {
            conf_map2.insert(st2, conf_map.get(&st2).unwrap());
        }
    }
    CONF.store(Some(Arc::new(conf_map2)));
    let no_dicts = get_no_reload_dicts(&dicts);
    if no_dicts.len() > 0 {
        return Err(anyhow::anyhow!(
            "存在没有配置到table-reload_dict_tables的字典表: {:?}",
            no_dicts
        )
        .into());
    }
    Ok(())
}

pub fn get_imp_config(st: SourceType) -> &'static Config {
    let conf_map = CONF.load().clone().unwrap();
    conf_map.get(&st).unwrap()
}

pub fn file_name_is_exist(fname: &str) -> bool {
    let conf_map = CONF.load().clone().unwrap();
    for (_, v) in conf_map.iter() {
        for t in v.tables.iter() {
            if t.file_name.as_str() == fname {
                return true;
            }
        }
    }
    false
}

fn get_no_reload_dicts(dicts: &Vec<(String, String)>) -> Vec<&str> {
    let conf_map = CONF.load().clone().unwrap();
    let mut res: Vec<&str> = Vec::new();
    for (dn, _) in dicts.iter() {
        let mut is_mh = false;
        for (_, v) in conf_map.iter() {
            for t in v.tables.iter() {
                if t.reload_dict_tables.contains(dn) {
                    is_mh = true;
                    break;
                }
            }
        }
        if !is_mh {
            res.push(dn);
        }
    }
    res
}
